package com.leon.raw_consumer.tool;


import com.leon.rpc.JSONSerializer;
import com.leon.rpc.RpcDecoder;
import com.leon.rpc.RpcEncoder;
import com.leon.rpc.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.I0Itec.zkclient.ZkClient;

import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class RpcConsumerEx {
    private static ExecutorService executor =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static List<UserClientHandler> clientHandlerList = new ArrayList<>();
    private static int index = 0;
    private static boolean refresh = false;

    private static String zkServer = "127.0.0.1:2181";
    private static String rootPath = "/raw_rpc";
    private static ZkClient zkClient;

    public RpcConsumerEx() {
        zkClient = new ZkClient(zkServer);
        try {
            initClients();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 针对每个远处服务创建一个代理对象（实际上应该创建一个单例的代理对象）
    public Object createProxy(final Class<?> serviceClass, final RpcRequest request) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                        new Class<?>[]{serviceClass}, (proxy, method, args) -> {

                    UserClientHandler client = chooseClientHandler();

                    request.setParameters(args);
                    request.setMethodName(method.getName());

                    // 填充参数类型数组，方便服务端根据方面名称和参数类型列表通过反射进行方法的定位和调用
                    Class[] types = new Class[args.length];
                    for (int i = 0; i < args.length; i++) {
                        types[i] = args[i].getClass();
                    }
                    request.setParameterTypes(types);
                    client.setRpcRequest(request);

                    long start = System.currentTimeMillis();
                    RpcRequest ret = (RpcRequest) executor.submit(client).get();
                    Object retResult = ret.getResult();
                    long end = System.currentTimeMillis();

                    long spend = end - start;

                    // 更新节点中响应时间以及上次请求时间
                    Object zkData = zkClient.readData(client.getZkNodePath());
                    String[] d = zkData.toString().split(",");
                    String newData = d[0] + "," + end + "," + spend;
                    zkClient.writeData(client.getZkNodePath(), newData);

                    return retResult;
                });
    }

    /**
     * 选择可用的 clientHandler (处理负载均衡)
     */
    private static UserClientHandler chooseClientHandler() throws Exception {

        while (refresh) {
            Thread.sleep(10);
        }

        if (clientHandlerList.size() == 0) {
            throw new Exception("No valid server");
        }

        // 当前负载均衡策略: 根据节点上的信息进行服务器筛选
        Map<Integer, Long> selection = new HashMap<>();
        long now = System.currentTimeMillis();

        int idx = -1;
        long repTime = 0;

        for (int i = 0; i < clientHandlerList.size(); i++) {
            Object data = zkClient.readData(clientHandlerList.get(i).getZkNodePath());
            String[] s = data.toString().split(",");
            long lastTime = Long.valueOf(s[1]).longValue();
            long lastResponseTime = Long.valueOf(s[2]).longValue();
            // 上次处理时间点超过 5 秒, 此 client 优先级提高
            if (now - lastTime > 5000) {
                selection.put(i, 0L);
            } else {
                selection.put(i, lastResponseTime);
            }
        }

        for (Map.Entry<Integer, Long> entry : selection.entrySet()) {
            if (idx == -1) {
                idx = entry.getKey().intValue();
                repTime = entry.getValue();
            } else if (entry.getValue() < repTime) {
                idx = entry.getKey();
                repTime = entry.getValue();
            }
        }
        UserClientHandler clientHandler = clientHandlerList.get(idx);

        /* 此处的负载均衡过程是轮询策略
        if (clientHandlerList.size() == 1) {
            index = 0;
        }else {
            index = index % clientHandlerList.size();
        }
        UserClientHandler clientHandler = clientHandlerList.get(index ++);
        */

        return clientHandler;
    }

    private static void doInitClient(String host, Integer port, String zkNodePath) {
        UserClientHandler client = new UserClientHandler();
        client.setZkNodePath(zkNodePath);
        clientHandlerList.add(client);

        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 500)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));
                        p.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        p.addLast(client);
                    } });

        try {
            ChannelFuture futrue =  b.connect(host, port).sync();
            //等待客户端链路关闭
//            futrue.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void initClients() throws Exception {
        System.out.println("first init UserClientHandler");
        boolean exists = zkClient.exists(rootPath);
        if (!exists) {
            throw new Exception("no valid server");
        }
        List<String> children = zkClient.getChildren(rootPath);
        children.stream().forEach(path -> processZKNodePath(rootPath + "/" + path));
        refresh = false;


        // 监听 rootPath 的子节点
        zkClient.subscribeChildChanges(rootPath, (s, list) -> {
            refresh = true;
            clientHandlerList.clear();

            System.out.println("active children node change listener");
            list.stream().forEach(path -> processZKNodePath(rootPath + "/" + path));
            refresh = false;

        });
    }

    private static void processZKNodePath(String nodePath) {
        Object o = zkClient.readData(nodePath);
        if (o != null) {
            String serverNodeInfo = o.toString();
            System.out.println("get server info :" + serverNodeInfo);

            String[] d = serverNodeInfo.split(",");

            // 初始化客户端连接
            String[] info = d[0].split(":");
            String host = info[0];
            Integer port = Integer.valueOf(info[1]);

            doInitClient(host, port, nodePath);
        }
    }

}
